package org.unigram.oluolu.recommend;

import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

/**
 *
 */
public class ExtractRecommendationReducer extends
        Reducer<LongWritable, Text, LongWritable, Text> {

    /**
     * reuduce method.
     * 
     * @param targetId target user id
     * @param recommends recommendation queries
     * @param context write down key and value 
     * 
     * @throws IOException -
     * @throws InterruptedException -
     */
    public final void reduce(final LongWritable targetId,
            final Iterable<Text> recommends, final Context context)
            throws IOException, InterruptedException {
        
        Map<String, Long> recommendMap = new HashMap<String, Long>();
        
        for (Text recommendText : recommends) {
            String recommendStr = recommendText.toString();
            String[] recommendAry = recommendStr.split(":");
            
            for (int i = 0; i< recommendAry.length; i++) {
                if (recommendMap.containsKey(recommendAry[i])) {
                    Long count = (Long) recommendMap.get(recommendAry[i]);
                    count += 1;
                    recommendMap.put(recommendAry[i], count);
                } else {
                    recommendMap.put(recommendAry[i], new Long(1));
                }
            }
        }
        
        Set<String> recommendSet = recommendMap.keySet();
        Iterator iterator = recommendSet.iterator();
        StringBuffer rtString = new StringBuffer(); 
        while (iterator.hasNext()) {
            String r = (String) iterator.next();
            if (recommendMap.get(r) > 3) {
                rtString.append(r);
                rtString.append(" ");
            }
        }
        
        context.write(targetId, new Text(rtString.toString()));
    }    
    
}
